package com.example.springboottest.example.rabbitmq.pubsub;

import com.example.springboottest.example.rabbitmq.RabbitUtil;
import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 *  1. 一个生产者，多个消费者；
 *  2. 每个消费者都有自己的队列；
 *  3. 生产者没有直接把消息发送到队列，而是发送到交换机，通过交换机转发到队列；
 * 4. 每个队列都要绑定到交换机上；
 *  5. 生产者发送的消息经过交换机到达队列就能实现一个消息被多个消费者消费；
 * 这里要注意：
 * 1. 在RabbitMQ中交换机没有存储能力，只有队列里面有；
 */
public class RabbitMQSub1 {
    private static final String QUEUE_NAME = "test_topic_queue_fanout_sms1";

    public static void main(String[] args) throws IOException, TimeoutException {

        Connection connection = RabbitUtil.getConnectionFactoryV2().newConnection();
        final Channel channel = connection.createChannel();

        //声明队列
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        //绑定到交换机
        channel.queueBind(QUEUE_NAME, RabbitMQPub.EXCHANGE_NAME, "");
        channel.basicQos(0);

        Consumer consumer = new DefaultConsumer(channel) {

            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {

                String msg = new String(body, "utf-8");
                System.out.println("[1] Recv msg : " + msg);

                try {

                    Thread.sleep(1000);
                } catch (InterruptedException e) {

                    e.printStackTrace();
                } finally {

                    System.out.println("[1] done");
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            }
        };

        boolean autoAck = false;
        channel.basicConsume(QUEUE_NAME, autoAck, consumer);
    }
}